/**
 * Copyright (C) FileName: MqttConfiguration Author:   a9714 Date:     2018/8/20 14:12 Description:
 * History:
 * <author>          <time>          <version>          <desc>
 * 作者姓名           修改时间           版本号              描述
 */
package com.haocang.filedemo.config;

import com.haocang.filedemo.handler.PasswordEmptyException;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

/**
 * 〈一句话功能简述〉<br>
 * 〈MQTT发送消息配置〉
 *
 * @author hugh
 * @create 2018/6/4
 * @since 1.0.0
 */
@Configuration
@IntegrationComponentScan
public class MqttSenderConfig {

  @Value("${mqtt.service.username}")
  private String username;

  @Value("${mqtt.service.password}")
  private String password;

  @Value("${mqtt.service.url}")
  private String hostUrl;

  @Value("${mqtt.client.appid}")
  private String clientId;

  @Value("${mqtt.service.qos}")
  private Integer qos;

  @Value("${mqtt.service.input-topic}")
  private String defaultTopic;

  public String getDefaultTopic() {
    return defaultTopic;
  }

  public void setDefaultTopic(String defaultTopic) {
    this.defaultTopic = defaultTopic;
  }

  public MqttConnectOptions getMqttConnectOptions() throws PasswordEmptyException {
    MqttConnectOptions mqttConnectOptions=new MqttConnectOptions();
    if (StringUtils.isNotBlank(username)) {
      mqttConnectOptions.setUserName(username);
      if (StringUtils.isNotBlank(password)) {
        mqttConnectOptions.setPassword(password.toCharArray());
      }else{
        throw new PasswordEmptyException("mqtt");
      }
    }
    // 设置mqtt服务地址，格式: 协议://ip:port
    mqttConnectOptions.setServerURIs(new String[]{hostUrl});
    // 设置是否清除session
    mqttConnectOptions.setCleanSession(false);
    // 设置连接存活时间
    mqttConnectOptions.setKeepAliveInterval(2);
    // 设置超时时间
    mqttConnectOptions.setConnectionTimeout(10);
    // 设置会话心跳时间
    mqttConnectOptions.setKeepAliveInterval(20);
    return mqttConnectOptions;
  }
  @Bean
  public MqttPahoClientFactory mqttClientFactory() throws PasswordEmptyException {
    DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
    factory.setConnectionOptions(getMqttConnectOptions());
    return factory;
  }

  @Bean
  @ServiceActivator(inputChannel = MqttConstant.OUTPUT_DATA_MQTT)
  public MessageHandler mqttOutbound() throws PasswordEmptyException {
    MqttPahoMessageHandler messageHandler =  new MqttPahoMessageHandler(clientId, mqttClientFactory());
    // 设置是否异步
    messageHandler.setAsync(true);
    // 设置发送topic
    messageHandler.setDefaultTopic(defaultTopic);
    // 被消费后是否保留当前数据
    messageHandler.setDefaultRetained(false);
    // 设置服务质量
    messageHandler.setDefaultQos(qos);
    return messageHandler;
  }

  @Bean
  public MessageChannel mqttInputChannel() {
    return new DirectChannel();
  }

  @Bean
  public MessageProducer getMessageProducer() throws PasswordEmptyException {
    MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("client_test",mqttClientFactory(),"Topic/haocang/mango/datacollect/inhand-test/Data");
    adapter.setConverter(new DefaultPahoMessageConverter());
    adapter.setQos(1);
    adapter.setOutputChannel(mqttInputChannel());
    return adapter;
  }

  @Bean
  @ServiceActivator(inputChannel = "mqttInputChannel")
  public MessageHandler getMessageHandler() {
    return new MyMessageHandler();
  }

}
